package com.nbcio.iot.service;

import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.nbcio.iot.domain.entity.IotDevice;
import com.nbcio.iot.mapper.IotDeviceMapper;
import com.nbcio.iot.protocol.modbus.ModbusManager;
import com.nbcio.iot.queue.IotMessageQueueService;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Service;

import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicBoolean;

import com.nbcio.iot.ingest.IotIngestionService;
import com.nbcio.common.tenant.helper.TenantHelper;
import com.nbcio.iot.mapper.IotProductMapper;

/**
 * IoT 数据采集服务
 * 定时采集设备数据并发送到消息队列
 *
 * @author nbacheng
 */
@Slf4j
@Service
@RequiredArgsConstructor
public class IotDataCollectionService {

    private final IotDeviceMapper deviceMapper;
    private final ModbusManager modbusManager;
    private final IotMessageQueueService messageQueueService;
    private final ObjectMapper objectMapper = new ObjectMapper();
    private final IotIngestionService ingestionService;
    private final NodeRedService nodeRedService;
    private final IotProductMapper productMapper;
    
    private final AtomicBoolean enabled = new AtomicBoolean(true);

    @Value("${iot.collection.enabled:true}")
    private boolean collectionEnabled;

    // 简单缓存：活跃且运行中的flowId集合，避免每5s都查库
    private volatile Set<Long> cachedActiveRunningFlows;
    private volatile long cachedAtMillis = 0L;
    // 默认缓存TTL 60秒，可通过配置覆盖
    @Value("${iot.collection.active-flow-cache-ttl-ms:60000}")
    private long activeFlowCacheTtlMs;

    // 新增：点位最后成功采集时间（内存级，单机生效；如需集群可改用Redis）
    private final Map<String, Long> lastPointSuccessTs = new ConcurrentHashMap<>();

    /**
     * 定时数据采集 - 每5秒执行一次
     */
    @Scheduled(fixedDelayString = "5000")
    public void collectData() {
        if (!enabled.get() || !collectionEnabled) {
            return;
        }
        
        try {
            Set<Long> activeRunningFlows = getCachedActiveRunningFlows();
            if (activeRunningFlows == null || activeRunningFlows.isEmpty()) {
                return;
            }

            // 一次查出绑定了这些规则的 modbus 设备
            List<IotDevice> devices = deviceMapper.selectList(
                new LambdaQueryWrapper<IotDevice>()
                    .in(IotDevice::getFlowId, activeRunningFlows)
                    .eq(IotDevice::getProtocolType, "modbus")
            );

            for (IotDevice device : devices) {
                if (!enabled.get()) {
                    break; // 如果服务被禁用，停止处理
                }
                
                try {
                    // 读取Modbus数据
                    Map<String, Object> data = readModbusData(device);
                    
                    // 在设备租户上下文中直接入库监控数据
                    TenantHelper.dynamic(device.getTenantId(), () -> {
                        int saved = ingestionService.saveMetrics(device, data);
                        if (saved > 0) {
                            log.info("[COLLECTION] metrics saved records={}, device={}", saved, device.getDeviceCode());
                        }
                    });
                    
                    // 构造消息（继续发送到规则引擎，不影响监控入库）
                    Map<String, Object> message = new HashMap<>();
                    message.put("tenantId", device.getTenantId());
                    message.put("deviceId", device.getDeviceId());
                    message.put("deviceCode", device.getDeviceCode());
                    message.put("productId", device.getProductId());
                    message.put("flowId", device.getFlowId());
                    message.put("payload", data);
                    message.put("timestamp", System.currentTimeMillis());
                    message.put("messageType", "DATA_COLLECTION");
                    
                    // 发送到消息队列
                    messageQueueService.sendDataCollectionMessage(message);
                    
                    log.info("[COLLECTION] 设备 {} 数据采集完成", device.getDeviceCode());
                } catch (Exception e) {
                    log.warn("[COLLECTION] 设备 {} 数据采集失败: {}", device.getDeviceCode(), e.getMessage());
                }
            }
        } catch (Exception e) {
            log.error("[COLLECTION] 数据采集服务异常: {}", e.getMessage());
        }
    }

    /**
     * 外部控制：停止数据采集服务
     */
    public void stop() {
        enabled.set(false);
        log.info("[COLLECTION] 数据采集服务已停止");
    }

    /**
     * 外部控制：启动数据采集服务
     */
    public void start() {
        enabled.set(true);
        log.info("[COLLECTION] 数据采集服务已启动");
    }

    private Set<Long> getCachedActiveRunningFlows() {
        long now = System.currentTimeMillis();
        if (cachedActiveRunningFlows != null && (now - cachedAtMillis) < activeFlowCacheTtlMs) {
            return cachedActiveRunningFlows;
        }
        // 超过TTL再查一次
        Set<Long> flows = nodeRedService.getActiveRunningFlowIds();
        cachedActiveRunningFlows = flows;
        cachedAtMillis = now;
        return flows;
    }

    /**
     * 读取Modbus数据（支持按点位的 periodSec 周期采集）
     */
    private Map<String, Object> readModbusData(IotDevice device) {
        Map<String, Object> data = new HashMap<>();
        try {
            Map<String, Object> config = parseDeviceConfig(device);
            if (config == null) return data;
            
            String type = (String) config.get("type");
            if ("tcp".equalsIgnoreCase(type)) {
                String host = (String) config.get("host");
                Integer port = (Integer) config.get("port");
                Integer timeoutMs = (Integer) config.get("timeoutMs");
                
                @SuppressWarnings("unchecked")
                List<Map<String, Object>> points = (List<Map<String, Object>>) config.get("points");
                if (points != null) {
                    long now = System.currentTimeMillis();
                    for (Map<String, Object> point : points) {
                        String code = (String) point.get("code");
                        Integer slaveId = asInt(point.get("slaveId"));
                        Integer offset = asInt(point.get("offset"));
                        Integer periodSec = asInt(point.get("periodSec")); // 可选

                        // 周期控制：未到时间则跳过
                        if (periodSec != null && periodSec > 0) {
                            String key = device.getDeviceId() + ":" + code;
                            Long last = lastPointSuccessTs.get(key);
                            if (last != null && now - last < periodSec * 1000L) {
                                continue;
                            }
                        }
                        
                        try {
                            Number value = modbusManager.readHoldingRegisterTcp(
                                host, port, timeoutMs != null ? timeoutMs : 2000, slaveId, offset
                            );
                            data.put(code, value);
                            // 成功则更新最后时间
                            if (periodSec != null && periodSec > 0) {
                                lastPointSuccessTs.put(device.getDeviceId() + ":" + code, now);
                            }
                        } catch (Exception e) {
                            log.warn("[COLLECTION] 读取点位 {} 失败: {}", code, e.getMessage());
                        }
                    }
                }
            }
        } catch (Exception e) {
            log.error("[COLLECTION] 解析设备配置失败: {}", e.getMessage());
        }
        return data;
    }

    private Integer asInt(Object v) {
        if (v == null) return null;
        if (v instanceof Number) return ((Number) v).intValue();
        try { return Integer.parseInt(String.valueOf(v)); } catch (Exception e) { return null; }
    }

    /**
     * 解析设备协议配置
     * 规则：先取产品protocol_config作为默认，再用设备protocol_config_override做覆盖；points按code合并，设备覆盖优先。
     */
    private Map<String, Object> parseDeviceConfig(IotDevice device) {
        try {
            Map<String, Object> base = new HashMap<>();
            // 产品默认
            if (device.getProductId() != null) {
                com.nbcio.iot.domain.entity.IotProduct prod = productMapper.selectById(device.getProductId());
                if (prod != null && prod.getProtocolConfig() != null && !prod.getProtocolConfig().isEmpty()) {
                    Map<String, Object> prodCfg = objectMapper.readValue(prod.getProtocolConfig(), Map.class);
                    if (prodCfg != null) base.putAll(prodCfg);
                }
            }
            // 设备覆盖（浅字段覆盖 + points按code合并）
            if (device.getProtocolConfigOverride() != null && !device.getProtocolConfigOverride().isEmpty()) {
                Map<String, Object> devCfg = objectMapper.readValue(device.getProtocolConfigOverride(), Map.class);
                if (devCfg != null) {
                    // 先处理points
                    @SuppressWarnings("unchecked")
                    List<Map<String, Object>> prodPoints = (List<Map<String, Object>>) base.get("points");
                    @SuppressWarnings("unchecked")
                    List<Map<String, Object>> devPoints = (List<Map<String, Object>>) devCfg.get("points");
                    if (devPoints != null) {
                        // 构建按code的合并
                        Map<String, Map<String, Object>> merged = new HashMap<>();
                        if (prodPoints != null) {
                            for (Map<String, Object> p : prodPoints) {
                                String code = String.valueOf(p.get("code"));
                                merged.put(code, new HashMap<>(p));
                            }
                        }
                        for (Map<String, Object> p : devPoints) {
                            String code = String.valueOf(p.get("code"));
                            Map<String, Object> exist = merged.getOrDefault(code, new HashMap<>());
                            exist.putAll(p); // 设备覆盖字段（含periodSec）
                            merged.put(code, exist);
                        }
                        base.put("points", merged.values().stream().toList());
                    }
                    // 其余顶层字段直接覆盖
                    for (Map.Entry<String, Object> e : devCfg.entrySet()) {
                        if (!"points".equals(e.getKey())) {
                            base.put(e.getKey(), e.getValue());
                        }
                    }
                }
            }
            return base.isEmpty() ? null : base;
        } catch (Exception e) {
            log.error("[COLLECTION] 解析设备配置失败: {}", e.getMessage());
            return null;
        }
    }
}
